package com.we.risk.antirefund.phoneinfo;

import com.alibaba.fastjson.JSONObject;
import com.we.flink.utils.WeKafkaPartitioner;
import com.we.flink.utils.WeKafkaPropertyReader;
import com.we.flink.utils.WeKafkaTopicSelector;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.sink.TopicSelector;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.*;

public class AdmBabelHfqPhoneInfoLabel {
    public static final String RELEASEPROP = "risk/antirefund/phoneinfo/adm_phone_info_kfk_prod.properties";
    public static Logger LOG = LoggerFactory.getLogger(AdmBabelHfqPhoneInfoLabel.class);

    public static final String KFKBROKERHOSTBAK =
            "10.10.13.22:9092,10.10.13.23:9092,10.10.13.24:9092";
    public static final String KFKBAKTOPIC = "kafka.topic.data.record";
    public static final String KFKBAKTOPICGRPID = "Babel_hfq_phont_info_grp_s2";

    public static final String MSGTYPE = "HFQ_UPLOAD_PHONE_INFO_RECORD";
    public static final String INPUTKEYBY = "userKey";
    public static final String OUTPUTKEY = "user_key";
    public static final String HTTPTIMESTAMP = "httpTimestamp";
    public static final String DEVICEID = "deviceId";
    public static final String OUTDEVICEID = "device_id";
    public static final String CELLIP = "cellIp";
    public static final String OUTCELLIP = "cell_ip";
    public static final String CELLIPCITY = "cellIPCity";
    public static final String OUTCELLIPCITY = "cell_ip_city";
    public static final String WIFIMAC = "wifiMac";
    public static final String OUTWIFIMAC = "wifi_mac";
    public static final String CAPTIME = "capture_time";

    public static final String SINKTOPIC = "sinkTopic";
    public static final String TOPICDEVICEID = "deviceid";
    public static final String TOPICCELLIP = "cellip";
    public static final String TOPICCELLIPCITY = "cellip_city";
    public static final String TOPICDWIFIMAC = "wifimac";

    public static final String SINKTOPICSUFFIX = "adm_anti_refund_phone_info_";
    public static final long RECVSTARTTIME = 1679932800L; //2023-03-28 00:00:00



    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();
            /** sink to Kafka */
            String sinkKfkAddress = paramReader.getKfkBootStrapServer();
            int sinkkfkPartitions = paramReader.getKfkPartitions();

            Properties consumProp = new Properties();
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            /** set offset to latest */
            KafkaSource<String> kfkSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(srcKafkaBootStrapServer)
                            .setTopics(srcKafkaTopic)
                            .setStartingOffsets(OffsetsInitializer.earliest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            consumProp.setProperty("group.id", KFKBAKTOPICGRPID);
            KafkaSource<String> kfkBakSource =
                    KafkaSource.<String>builder()
                            .setBootstrapServers(KFKBROKERHOSTBAK)
                            .setTopics(KFKBAKTOPIC)
                            .setStartingOffsets(OffsetsInitializer.latest())
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            SingleOutputStreamOperator<String> input =
                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "phoneInfo")
                            .uid("kfk-phoneinfo-source");
            SingleOutputStreamOperator<String> inputBak =
                    env.fromSource(kfkBakSource, WatermarkStrategy.noWatermarks(), "phoneInfo-bak")
                            .uid("kfk-phoneinfo-bak-source");
            DataStream<String> unionInput = input.union(inputBak);

            SingleOutputStreamOperator<String> out = unionInput.flatMap(new phoneInfoRichFlatMapFunction());

            KafkaRecordSerializationSchema<String> serializedRecord =
                    KafkaRecordSerializationSchema.builder()
                            .setTopicSelector(
                                    new phoneInfoTopicSelector())
                            .setKeySerializationSchema(
                                    new SerializationSchema<String>() {
                                        @Override
                                        public byte[] serialize(String element) {
                                            JSONObject input = JSONObject.parseObject(element);
                                            String key = input.getString(OUTPUTKEY);
                                            return key.getBytes();
                                        }
                                    })
                            .setValueSerializationSchema(
                                    new SerializationSchema<String>() {
                                        @Override
                                        public byte[] serialize(String element) {
                                            return element.getBytes();
                                        }
                                    })
                            .setPartitioner(new WeKafkaPartitioner())
                            .build();

            KafkaSink<String> kfkSink =
                    KafkaSink.<String>builder()
                            .setBootstrapServers(sinkKfkAddress)
                            .setRecordSerializer(serializedRecord)
                            .build();

            out.sinkTo(kfkSink).setParallelism(sinkkfkPartitions);

            env.execute(AdmBabelHfqPhoneInfoLabel.class.toString());
        } catch (Exception e) {
            System.out.println("Exception: " + e.getMessage());
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class phoneInfoTopicSelector implements TopicSelector<String> {
        @Override
        public String apply(String s) {
            JSONObject inputJson = JSONObject.parseObject(s);
            String sinkTopic = inputJson.getString(SINKTOPIC);
            LOG.debug("SinkTopic: " + sinkTopic);
            return sinkTopic;
        }
    }


    private static class phoneInfoRichFlatMapFunction extends RichFlatMapFunction<String, String> {
        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            try {
                HfqPhoneInfo hfqPhoneInfo = filterphoneInfoMsg(value);
                LOG.debug("hfqPhoneInfo: " + hfqPhoneInfo.toString());
                String user_key = hfqPhoneInfo.getUser_key();
                long cap_time = hfqPhoneInfo.getCap_time();
                if(cap_time != 0 && cap_time >= RECVSTARTTIME) {
                    String device_id = hfqPhoneInfo.getDevice_id();
                    String cell_ip = hfqPhoneInfo.getCell_ip();
                    String cell_ip_city = hfqPhoneInfo.getCell_ip_city();
                    String wifi_mac = hfqPhoneInfo.getWifi_mac();
                    if (user_key != null) {
                        JSONObject outJson = new JSONObject();
                        outJson.put(OUTPUTKEY, user_key);
                        outJson.put(CAPTIME, cap_time);

                        //deviceID
                        if (device_id != null && !"".equals(device_id)) {
                            outJson.put(SINKTOPIC, SINKTOPICSUFFIX + TOPICDEVICEID);
                            outJson.put(OUTDEVICEID, device_id);
                            LOG.debug("deviceID outJson: " + outJson.toString());
                            out.collect(outJson.toString());
                        }

                        //wifiMac
                        if (wifi_mac != null && !"".equals(wifi_mac)
                                && !"null".equals(wifi_mac) && !"00:00:00:00:00:00".equals(wifi_mac)
                                && !"02:00:00:00:00:00".equals(wifi_mac)) {
                            outJson.put(SINKTOPIC, SINKTOPICSUFFIX + TOPICDWIFIMAC);
                            outJson.remove(OUTDEVICEID);
                            outJson.put(OUTWIFIMAC, wifi_mac);
                            LOG.debug("wifiMac outJson: " + outJson.toString());
                            out.collect(outJson.toString());
                        }

                        //ip
                        if (cell_ip != null && !"".equals(cell_ip)) {
                            outJson.put(SINKTOPIC, SINKTOPICSUFFIX + TOPICCELLIP);
                            outJson.remove(OUTDEVICEID);
                            outJson.remove(OUTWIFIMAC);
                            outJson.put(OUTCELLIP, cell_ip);
                            LOG.debug("ip outJson: " + outJson.toString());
                            out.collect(outJson.toString());
                        }

                        //ipCity
                        if (cell_ip_city != null && !"".equals(cell_ip_city)) {
                            outJson.put(SINKTOPIC, SINKTOPICSUFFIX + TOPICCELLIPCITY);
                            outJson.remove(OUTCELLIP);
                            outJson.remove(OUTDEVICEID);
                            outJson.remove(OUTWIFIMAC);
                            outJson.put(OUTCELLIPCITY, cell_ip_city);
                            LOG.debug("ipCity outJson: " + outJson.toString());
                            out.collect(outJson.toString());
                        }
                    } else {
                        LOG.warn("phoneInfoRichFlatMapFunction UserKey or Captime error!!");
                    }
                }
            } catch (Exception e) {
                LOG.error(e.getMessage());
                LOG.error("phoneInfoRichFlatMapFunction Exception input: " + value);
            }
        }

    }

    private static HfqPhoneInfo filterphoneInfoMsg(String value) {
        HfqPhoneInfo hfqPhoneInfo = new HfqPhoneInfo();
        try {
            if (value != null && value.contains(MSGTYPE)) {
                JSONObject inputJson = JSONObject.parseObject(value);
                if (inputJson != null) {
                    String type = inputJson.getString("type");
                    if (type != null && type.equals(MSGTYPE)) {
                        JSONObject datajson = inputJson.getJSONObject("dataJson");
                        if (datajson != null) {
                            String userKey = datajson.getString(INPUTKEYBY);
                            hfqPhoneInfo.setUser_key(userKey);

                            JSONObject data = datajson.getJSONObject("data");
                            if (data != null) {
                                String deviceId = data.getString(DEVICEID);
                                hfqPhoneInfo.setDevice_id(deviceId);
                                String cellIp = data.getString(CELLIP);
                                hfqPhoneInfo.setCell_ip(cellIp);
                                String cellIpCity = data.getString(CELLIPCITY);
                                hfqPhoneInfo.setCell_ip_city(cellIpCity);
                                String wifiMac = data.getString(WIFIMAC);
                                hfqPhoneInfo.setWifi_mac(wifiMac);

                                JSONObject baseData = data.getJSONObject("baseData");
                                if (baseData != null) {
                                    long capture_time = baseData.getLongValue(HTTPTIMESTAMP);
                                    hfqPhoneInfo.setCap_time(capture_time);
                                }
                            }
                        }
                    }
                }
            }
        } catch (Exception e) {
            LOG.error(e.getMessage());
            LOG.error("filterSmsInfoMsg excep input: " + value);
        }
        return hfqPhoneInfo;
    }


    private static JSONObject initPropJson(String propFile) throws IOException {
        InputStream resourceAsStream =
                AdmBabelHfqPhoneInfoLabel.class.getClassLoader().getResourceAsStream(propFile);
        ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
        JSONObject res = new JSONObject();
        Iterator<Object> keysIterator = reader.getProperties().keySet().iterator();
        while (keysIterator.hasNext()) {
            String key = (String) keysIterator.next();
            if (key.equals(CAPTIME)) {
                res.put(key, -1);
            } else {
                res.put(key, "initV");
            }
        }
        return res;
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getString(OUTPUTKEY);
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.debug(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }
}
